package com.erainm.dw.flume.interceptor;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.compress.utils.Charsets;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.event.SimpleEvent;
import org.apache.flume.interceptor.Interceptor;
import org.junit.Test;

import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * Created by erainm on 2021/9/19 3:41 下午.
 *
 * @Description TODO:
 **/
public class CustomerInterceptor implements Interceptor {
    @Override
    public void initialize() {

    }

    /**
     * Created by erainm on 2021/9/19 3:43 下午.
     * @Param [event]
     * @return [org.apache.flume.Event]
     * @Description TODO: 逐条处理event
     **/
    @Override
    public Event intercept(Event event) {
//        1、获取 event 的 body
        String enentBody = new String(event.getBody(), Charsets.UTF_8);
//        2、获取 event 的 header
        Map<String, String> headersMap = event.getHeaders();
//        3、解析body获取json串
        String[] bodyArr = enentBody.split("\\s+");
        try {
            String jsonStr = bodyArr[6];
            // 4、解析json串获取时间戳
            JSONObject jsonObject = JSON.parseObject(jsonStr);
            String timestamStr = jsonObject.getJSONObject("app_active").getString("time");
            // 5、将时间戳转换为字符串 "yyyy-MM-dd"
            long timestamp = Long.parseLong(timestamStr);
            DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd");
            Instant instant = Instant.ofEpochMilli(timestamp);
            LocalDateTime localDateTime = LocalDateTime.ofInstant(instant, ZoneId.systemDefault());
            String date = formatter.format(localDateTime);
            // 6、将转换后的字符串放置header中
            headersMap.put("logtime",date);
            event.setHeaders(headersMap);
            // 7、返回event
        } catch (Exception e) {
            headersMap.put("logtime","UnKnow");
            event.setHeaders(headersMap);
        }

        return event;
    }

    @Override
    public List<Event> intercept(List<Event> events) {
        List<Event> listEvent = new ArrayList<>();
        for (Event event : events) {
            Event outEvent = intercept(event);
            if (null != outEvent)
                listEvent.add(outEvent);
        }
        return listEvent;
    }

    @Override
    public void close() {

    }

    public static class Builder implements Interceptor.Builder {

        @Override
        public Interceptor build() {
            return new CustomerInterceptor();
        }

        @Override
        public void configure(Context context) {

        }
    }

    @Test
    public void testJunit(){
        String str = "2020-08-20 11:56:08.698 [main] INFO  com.lagou.ecommerce.AppStart - {\"app_active\":{\"name\":\"app_active\",\"json\":{\"entry\":\"2\",\"action\":\"0\",\"error_code\":\"0\"},\"time\":1595295367796},\"attr\":{\"area\":\"厦门\",\"uid\":\"2F10092A9733\",\"app_v\":\"1.1.16\",\"event_type\":\"common\",\"device_id\":\"1FB872-9A1009733\",\"os_type\":\"0.5.0\",\"channel\":\"UZ\",\"language\":\"chinese\",\"brand\":\"Huawei-6\"}}";
        Map<String,String> map = new HashMap<>();
        // 创建一个Event
        SimpleEvent event = new SimpleEvent();
        event.setHeaders(map);
        event.setBody(str.getBytes(Charsets.UTF_8));

        // 调用Interceptor处理event
        CustomerInterceptor customerInterceptor = new CustomerInterceptor();
        Event outEvent = customerInterceptor.intercept(event);
        // 处理的结果
        Map<String, String> headersMap = outEvent.getHeaders();
        System.out.println(JSON.toJSONString(JSON.toJSONString(headersMap)));
    }
}
